package cn.itedus.lottery.application.mq.consumer;

import cn.itedus.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
import cn.itedus.lottery.domain.activity.service.partake.IActivityPartake;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author aodre , QQ : 480029069
 * @date 2023/10/5 17:04
 */
@Component
@RocketMQMessageListener(topic = "lottery_activity_partake",consumerGroup = "lottery-stock")
public class RocketLotteryActivityPartakeRecordListener implements RocketMQListener<MessageExt> {


    private Logger logger = LoggerFactory.getLogger(LotteryActivityPartakeRecordListener.class);

    @Resource
    private IActivityPartake activityPartake;
    /*
     TODO: 可以实现,批量拉去MQ 中的消息, 然后批量更新 Mysql ,相当于把多个sql 合并成一个sql ,批量更新的 幂等处理需要考虑一下
     */
    @Override
    public void onMessage(MessageExt message) {

        // 0. 实际场景在消费MQ的时候，可以通过固定标识判断是否已经消费过，添加记录。对程序的处理会起到优化作用。

        // 1. 判断消息是否存在
        if (message == null) {
            return;
        }
        ActivityPartakeRecordVO activityPartakeRecordVO = null;
        try {
            // 2. 转化对象（或者你也可以重写Serializer<T>）
            activityPartakeRecordVO = JSON.parseObject(new String(message.getBody()), ActivityPartakeRecordVO.class);
            logger.info("消费MQ消息，异步扣减活动库存 message：{}", activityPartakeRecordVO);

            // 3. 更新数据库库存【实际场景业务体量较大，可能也会由于MQ消费引起并发，对数据库产生压力，所以如果并发量较大，可以把库存记录缓存中，并使用定时任务进行处理缓存和数据库库存同步，减少对数据库的操作次数】
            activityPartake.updateActivityStock(activityPartakeRecordVO);
        }catch (Exception e){
            logger.info("消费MQ消息失败 -----> message：{}", activityPartakeRecordVO);
            throw  e;
        }

    }
}
